package com.eshipenet.shipnet.equipmentextbuilder

import org.apache.flink.streaming.api.scala._
import play.api.libs.json.{JsDefined, JsObject, JsString, JsValue, Json}

object EquipmentWithModelActionStreamTest {

  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    val equipmentWithModelIdActionStream: DataStream[EquipmentWithModelIdAction] =
      createEquipmentWithModelIdActionStream(env)

    val equipmentModelActionStream: DataStream[EquipmentModelAction] = createEquipmentModelActionStream(env)
    val s =
      Processor.generateEquipmentWithModelActionStream(equipmentWithModelIdActionStream, equipmentModelActionStream)
    s.print()
    env.execute()
  }

  def createEquipmentWithModelIdActionStream(
      env: StreamExecutionEnvironment): DataStream[EquipmentWithModelIdAction] = {
    env
      .socketTextStream("0.0.0.0", 4001, '\n', 10)
      .map(s => Json.parse(s))
      .map { json =>
        json \ "op" match {
          case JsDefined(JsString("i")) =>
            val raw       = (json \ "equipment").as[JsObject]
            val equipment = Equipment(raw).get
            val modelId   = (json \ "modelId").as[String]
            val link      = EquipmentModelLink(equipment.id, modelId, Map(), Json.obj())
            Some(EquipmentWithModelIdInsert(EquipmentWithLink(equipment, link)))
          case JsDefined(JsString("d")) =>
            val raw       = (json \ "equipment").as[JsObject]
            val equipment = Equipment(raw).get
            val modelId   = (json \ "modelId").as[String]
            val link      = EquipmentModelLink(equipment.id, modelId, Map(), Json.obj())
            Some(EquipmentWithModelIdDelete(EquipmentWithLink(equipment, link)))
          case _ => None
        }
      }
      .filter(_.isDefined)
      .map(_.get)
  }

  def createEquipmentModelActionStream(env: StreamExecutionEnvironment): DataStream[EquipmentModelAction] = {
    env
      .socketTextStream("0.0.0.0", 4002, '\n', 10)
      .map(s => Json.parse(s))
      .map { json =>
        json \ "op" match {
          case JsDefined(JsString("i")) =>
            val raw   = (json \ "model").as[JsObject]
            val model = EquipmentModel(raw).get
            Some(EquipmentModelInsert(model.id, model))
          case JsDefined(JsString("d")) =>
            val raw   = (json \ "model").as[JsObject]
            val model = EquipmentModel(raw).get
            Some(EquipmentModelDelete(model.id, model))
          case _ => None
        }
      }
      .filter(_.isDefined)
      .map(_.get)
  }

}
